#pragma once
#include <cstdint>
#include <functional>
#include <memory>
#include "Channel.hpp"
#include "Common/thread_pool.hpp"
#include "Connection.hpp"
#include "MQ_Virtual_Engine/VirtualHost.hpp"
#include "codec/codec.h"
#include "codec/dispatcher.h"
#include "fmtlog/fmtlog.h"
#include "google/protobuf/message.h"
#include "mq_proto.pb.h"
#include "muduo/base/Timestamp.h"
#include "muduo/net/Callbacks.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/InetAddress.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/net/TcpServer.h"
#include "Comsumer.hpp"

namespace MY_MQ {

#define DBFILE   "/test.db"
#define HOSTNAME "MyVirtualHost"

class Server {
public:
    using MessagePtr = std::shared_ptr<google::protobuf::Message>;

    Server(int16_t port, const std::string& basidir)
        : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort),
          _dispatcher(std::bind(&Server::onUnkownMessage, this, std::placeholders::_1, std::placeholders::_2,
                                std::placeholders::_3)),
          _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                                                           std::placeholders::_1, std::placeholders::_2,
                                                           std::placeholders::_3))),
          _vitualHost(std::make_shared<VirtualHost>(HOSTNAME, basidir, basidir + DBFILE)),
          _consumer_manager(std::make_shared<ConsumerManager>()),
          _connection_manager(std::make_shared<ConnectionManager>()),
          _thread_pool(std::make_shared<Message_MQ::ThreadPool>()) {
        auto qm = _vitualHost->AllQueues();
        for (auto& q : qm) {
            _consumer_manager->InitComsumerQueue(q.first);
        }
        _dispatcher.registerMessageCallback<MQ_Protobuf::openChannelRequest>(std::bind(
            &Server::onOpenChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<MQ_Protobuf::closeChannelRequest>(std::bind(
            &Server::onCloseChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<MQ_Protobuf::declareExchangeRequest>(std::bind(
            &Server::onDeclareExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<MQ_Protobuf::deleteExchangeRequest>(std::bind(
            &Server::onDeleteExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<MQ_Protobuf::declareQueueRequest>(std::bind(
            &Server::onDeclareQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

        _dispatcher.registerMessageCallback<MQ_Protobuf::deleteQueueRequest>(std::bind(
            &Server::onDeleteQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<MQ_Protobuf::queueBindRequest>(
            std::bind(&Server::onQueueBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<MQ_Protobuf::queueUnBindRequest>(std::bind(
            &Server::onQueueUnBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<MQ_Protobuf::basicPublishRequest>(std::bind(
            &Server::onBasicPublish, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<MQ_Protobuf::basicAckRequest>(
            std::bind(&Server::onBasicAck, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<MQ_Protobuf::basicConsumeRequest>(std::bind(
            &Server::onBasicConsume, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<MQ_Protobuf::basicCancelRequest>(std::bind(
            &Server::onBasicCancel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec, std::placeholders::_1,
                                             std::placeholders::_2, std::placeholders::_3));
        _server.setConnectionCallback(std::bind(&Server::onConnection, this, std::placeholders::_1));
    }

    void start() {
        _server.start();
        _baseloop.loop();
    }

private:
    void onConnection(const muduo::net::TcpConnectionPtr& conn) {
        if (conn->connected()) {
            _connection_manager->newConnection(_vitualHost, _consumer_manager, _codec, conn, _thread_pool);
        } else {
            _connection_manager->DeleteConnection(conn);
        }
    }

    void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& Message,
                       muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("打开信道时不存在对应的Connection对象!\n");
            conn->shutdown();
            return;
        }
        m_conn->OpenChannel(Message);
        logd("信道创建成功{}",Message->cid());
    }

    void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& message,
                        muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("关闭信道时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        return m_conn->closeChannel(message);
    }

    void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& message,
                           muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("声明交换机时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("声明交换机时，没找到对应的Channel对象");
            return;
        }

        return cp->DeclareExchange(message);
    }

    void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& message,
                          muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("删除交换机时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("删除交换机时，没找到对应的Channel对象");
            return;
        }

        return cp->deleteExchange(message);
    }

    void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& message,
                        muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("声明队列时，没找到对应的Connection对象");
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("声明队列时，没找到对应的Channel对象");
        }
        return cp->DeclareQueue(message);
    }

    void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& message,
                       muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("删除队列时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("删除队列时，没找到对应的Channel对象");
            return;
        }

        return cp->DeleteQueue(message);
    }

    void onQueueBind(const muduo::net::TcpConnectionPtr& conn, const queueBindRequestPtr& message, muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("绑定队列时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("绑定队列时，没找到对应的Channel对象");
            return;
        }

        return cp->queueBind(message);
    }

    void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn, const queueUnBindRequestPtr& message,
                       muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("解绑队列时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("解绑队列时，没找到对应的Channel对象");
            return;
        }
        cp->queueUbind(message);
    }

    void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& message,
                        muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("发布消息时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("发布消息时，没找到对应的Channel对象");
            return;
        }
        cp->basicPublish(message);
    }

    void onBasicAck(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& message, muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("确认消息时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("确认消息时，没找到对应的Channel对象");
            return;
        }
        cp->basicAck(message);
    }

    void onBasicConsume(const muduo::net::TcpConnectionPtr& conn, const basicConsumeRequestPtr& message,
                        muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("消费消息时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("消费消息时，没找到对应的Channel对象");
            return;
        }
        cp->basicConsume(message);
    }

    void onBasicCancel(const muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& message,
                       muduo::Timestamp) {
        Connection::self_ptr m_conn = _connection_manager->getConnection(conn);
        if (nullptr == m_conn) {
            logd("取消消费消息时，没找到对应的Connection对象");
            conn->shutdown();
            return;
        }
        Channel::self_ptr cp = m_conn->getChannel(message->cid());
        if (nullptr == cp) {
            logd("取消消费消息时，没找到对应的Channel对象");
            return;
        }
        cp->basicCancel(message);
    }

    void onUnkownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) {
        logd("UnKownMessage：{}", message->GetTypeName());
        conn->shutdown();
    }

private:
    muduo::net::EventLoop            _baseloop;
    muduo::net::TcpServer            _server;
    ProtobufDispatcher               _dispatcher;
    ProtobufCodecPtr                 _codec;
    VirtualHost::self_ptr            _vitualHost;
    ConsumerManager ::self_ptr       _consumer_manager;
    ConnectionManager::self_ptr      _connection_manager;
    Message_MQ::ThreadPool::self_ptr _thread_pool;
};
};  // namespace MY_MQ
